package com.wangzhen.exchangesTwo.config;

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;
import java.util.UUID;

/**
 * @author wz
 * @company 公司
 * @create 2020-04-14 16:46
 */
public class BusinessMessageReceiver implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback, InitializingBean {
    public static final Logger log = LoggerFactory.getLogger(BusinessMessageReceiver.class);

    @Autowired
    private RabbitTemplate rabbitTemplate;

    //监听实时数据队列
    @RabbitListener(queues = RabbitMQConfig.BUSINESS_REALTIEMMSG_NAME)
    public void receiveRealTimeMsg(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("收到othermsg：{}", msg);
        System.out.println(message.getMessageProperties().getDeliveryTag());
        boolean ack = true;
        Exception exception = null;
        try {
            //处理业务逻辑
        } catch (Exception e){
            ack = false;
            exception = e;
        }
        //判断ack是否报错则加入死信队列
        if (ack){
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        } else {
            log.error("消息消费发生异常，error msg:{}", exception.getMessage(), exception);
            //拒收报错信息，并将报错信息加入延迟队列，延迟一定
            /**
             * 上面的代码中，这句算是踢出当前队列的。也就是踢出主队列，因为主队列的数据量非常大，所以把报错的或者拒收的都放到延时队列中等时间到了 以后转入死信队列进行再次消费投递。如果在失败就进行相应的保存处理。
             */
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);

            //加入延迟队列
            sendDelayQueue(new String(message.getBody()),"iov.delay.othermsg");
        }
    }

    public void sendDelayQueue(String msg,String routingKey){

        Message message = MessageBuilder.withBody(msg.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("utf-8")
                .setMessageId(UUID.randomUUID()+"").build();
        // 自定义消息唯一标识
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(message.getMessageProperties().getMessageId());

        rabbitTemplate.convertSendAndReceive(RabbitMQConfig.BUSINESS_DELAY_EXCHANGE_NAME, routingKey, message,correlationData);

    }


    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {

    }

    @Override
    public void returnedMessage(Message message, int i, String s, String s1, String s2) {

    }

    @Override
    public void afterPropertiesSet() throws Exception {

    }
}
